/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package consys.social.nosql.service.hbase;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

/**
 *
 * @author petan
 */
public class HBaseHelpers {

    // HBase objects creaters
    public static Put createPut(String row) {
        return new Put(Bytes.toBytes(row));
    }

    public static Get createGet(String row) {
        return new Get(Bytes.toBytes(row));
    }

    public static Delete createDelete(String row) {
        return new Delete(Bytes.toBytes(row));
    }

    public static Scan createScan(String from, String to) {
        return new Scan(Bytes.toBytes(from), Bytes.toBytes(to));
    }

    public static Scan createScan(String from) {
        return new Scan(Bytes.toBytes(from));
    }

    public static Scan createScan() {
        return new Scan();
    }

    // HBase Put helpers
    public static void addToPut(Put put, String cf, String col, byte[] data) {
        put.add(Bytes.toBytes(cf), Bytes.toBytes(col), data);
    }

    public static void addToPut(Put put, String cfCol, byte[] bytes) {
        String[] colFamiliesAndColumns = HBaseUtils.splitFamilyAndCol(cfCol);
        addToPut(put, colFamiliesAndColumns[0], colFamiliesAndColumns[1], bytes);
    }

    public static void addToPut(Put put, String cf, String col, String s) {
        addToPut(put, cf, col, Bytes.toBytes(s));
    }

    public static <T> void addToPut(Put put, String cf, String col, T o, ByteTranscoder<T> t) {
        addToPut(put, cf, col, t.encode(o));
    }

    public static void addToPut(Put put, String cf, String col, int i) {
        addToPut(put, cf, col, Bytes.toBytes(i));
    }

    public static void addToPut(Put put, String cf, String col, long l) {
        addToPut(put, cf, col, Bytes.toBytes(l));
    }

    public static void addToPut(Put put, String cf, String col, boolean b) {
        addToPut(put, cf, col, Bytes.toBytes(b));
    }

    public static void addToPut(Put put, String cfCol, String s) {
        addToPut(put, cfCol, Bytes.toBytes(s));
    }

    public static void addToPut(Put put, String cfCol, int i) {
        addToPut(put, cfCol, Bytes.toBytes(i));
    }

    public static void addToPut(Put put, String cfCol, long l) {
        addToPut(put, cfCol, Bytes.toBytes(l));
    }

    public static void addToPut(Put put, String cfCol, boolean b) {
        addToPut(put, cfCol, Bytes.toBytes(b));
    }

    public static void addMoreToPut(Put put, String[] identifiers, byte[][] bytes) {
        if (identifiers.length != bytes.length) {
            //log.error("array of keys doesnt have same length as array of values");
            throw new IllegalArgumentException("Array of keys doesnt have same length as array of values");
        }
        for (int i = 0; i < identifiers.length; i++) {
            addToPut(put, identifiers[i], bytes[i]);
        }
    }

    // HBase Get helpers

    public static void addToGet(Get get, String cf, String col) {
        get.addColumn(Bytes.toBytes(cf), Bytes.toBytes(col));
    }

    public static void addFamilyToGet(Get get, String cf) {
        get.addFamily(Bytes.toBytes(cf));
    }

    public static void addToGet(Get get, String cfCol) {
        String[] colFamiliesAndColumns = HBaseUtils.splitFamilyAndCol(cfCol);
        addToGet(get, colFamiliesAndColumns[0], colFamiliesAndColumns[1]);
    }

    public static void addMoreToGet(Get get, String[] identifiers) {
        for (int i = 0; i < identifiers.length; i++) {
            addToGet(get, identifiers[i]);
        }
    }

    // HBase Delete helpers
    public static void deleteCol(Delete del, String cf, String col) {
        del.deleteColumns(Bytes.toBytes(cf), Bytes.toBytes(col));
    }

    public static void deleteCol(Delete del, String cfCol) {
        String[] colFamiliesAndColumns = HBaseUtils.splitFamilyAndCol(cfCol);
        deleteCol(del, colFamiliesAndColumns[0], colFamiliesAndColumns[1]);
    }

    public static void deleteFamily(Delete del, String cf) {
        del.deleteFamily(Bytes.toBytes(cf));
    }

    // HBase Scan helpers
    public static void addToScan(Scan scan, String cf, String col) {
        scan.addColumn(Bytes.toBytes(cf), Bytes.toBytes(col));
    }

    public static void addToScan(Scan scan, String cfCol) {
        String[] colFamiliesAndColumns = HBaseUtils.splitFamilyAndCol(cfCol);
        addToScan(scan, colFamiliesAndColumns[0], colFamiliesAndColumns[1]);
    }

    public static void addMoreToScan(Scan scan, String[] identifiers) {
        for (int i = 0; i < identifiers.length; i++) {
            addToScan(scan, identifiers[i]);
        }
    }

    public static void addFamilyToScan(Scan scan, String cf) {
        scan.addFamily(Bytes.toBytes(cf));
    }

    public static void addFamiliesToScan(Scan scan, String[] identifiers) {
        for (int i = 0; i < identifiers.length; i++) {
            addFamilyToScan(scan, identifiers[i]);
        }
    }

    // HBase Result helpers

    public static byte[] getValue(Result r, String cf, String col) {
        return r.getValue(Bytes.toBytes(cf), Bytes.toBytes(col));
    }

    public static byte[] getValue(Result r, String cfCol) {
        String[] colFamiliesAndColumns = HBaseUtils.splitFamilyAndCol(cfCol);
        return getValue(r, colFamiliesAndColumns[0], colFamiliesAndColumns[1]);
    }

    public static boolean containsColumn(Result r, String family, String col) {
        return r.containsColumn(Bytes.toBytes(family), Bytes.toBytes(col));
    }

    public static byte[] getColumnValue(Result r, String family, String col) {
        return r.getColumnLatest(Bytes.toBytes(family), Bytes.toBytes(col)).getValue();
    }
}
